package com.sniper.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * Created by sniper on 2017/6/25.
 */
public class HBaseHolder {


    Configuration conf = HBaseConfiguration.create();
    Connection conn;
    String tableName = "sniper_2";

    public HBaseHolder() throws IOException {

        conf.set("hbase.zookeeper.quorum", "master:2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    public void createTable(String tableName) throws IOException {

        TableName _tableName = TableName.valueOf(tableName);
        Admin admin = conn.getAdmin();
        if (admin.tableExists(_tableName)) {
            System.out.println("表已存在");
            return;
        }

        HTableDescriptor desc = new HTableDescriptor(_tableName);
        desc.addFamily(new HColumnDescriptor("cf1"));
        desc.addFamily(new HColumnDescriptor("cf2"));
        admin.createTable(desc);
    }

    /*
    * 为表添加数据（适合知道有多少列族的固定表）
    *
    * @rowKey rowKey
    *
    * @tableName 表名
    *
    * @column1 第一个列族列表
    *
    * @value1 第一个列的值的列表
    *
    * @column2 第二个列族列表
    *
    * @value2 第二个列的值的列表
    */
    public void addData(String rowKey, String tableName, String[] column1, String[] value1, String[] column2, String[] value2)
            throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));// 设置rowkey
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        // HTabel负责跟记录相关的操作如增删改查等//

        // 获取表
        // 获取所有的列族
        HColumnDescriptor[] columnFamilies = table.getTableDescriptor().getColumnFamilies();

        for (HColumnDescriptor columnFamily : columnFamilies) {
            String familyName = columnFamily.getNameAsString(); // 获取列族名
            System.out.println(familyName);
            if (familyName.equals("cf1")) { // article列族put数据
                for (int j = 0; j < column1.length; j++) {
                    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column1[j].trim()), Bytes.toBytes(value1[j].trim()));
                }
            }
            if (familyName.equals("cf2")) { // author列族put数据
                for (int j = 0; j < column2.length; j++) {
                    put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(column2[j].trim()), Bytes.toBytes(value2[j].trim()));
                }
            }
        }
        table.put(put);
        table.close();
        System.out.println("add data Success!");
    }

    public Result getResult(String tableName, String rowKey) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        System.out.println(cells.size());
        for (Cell cell : cells) {
            System.out.println("family:" + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            System.out.println("qualifier:" + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            System.out.println("value:" + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            System.out.println("Timestamp:" + cell.getTimestamp());
            System.out.println("-------------------------------------------");
        }
        return result;
    }

    public void getResultScan(String tableName) throws IOException {
        Scan scan = new Scan();
        ResultScanner rs = null;
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        try {
            rs = table.getScanner(scan);
            for (Result r : rs) {
                for (Cell kv : r.listCells()) {
                    System.out.println("row:" + Bytes.toString(kv.getRowArray()));
                    System.out.println("family:" + Bytes.toString(kv.getFamilyArray()));
                    System.out.println("qualifier:" + Bytes.toString(kv.getQualifierArray()));
                    System.out.println("value:" + Bytes.toString(kv.getValueArray()));
                    System.out.println("timestamp:" + kv.getTimestamp());
                    System.out.println("-------------------------------------------");
                }
            }
        } finally {
            rs.close();
        }
    }

    /*
    * 遍历查询hbase表
    *
    * @tableName 表名
    */
    public void getResultScan(String tableName, String start_rowkey, String stop_rowkey) throws IOException {
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(start_rowkey));
        scan.setStopRow(Bytes.toBytes(stop_rowkey));
        ResultScanner rs = null;
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        try {
            rs = table.getScanner(scan);
            for (Result r : rs) {
                for (Cell kv : r.listCells()) {
                    System.out.println("row:" + Bytes.toString(kv.getRowArray()));
                    System.out.println("family:" + Bytes.toString(kv.getFamilyArray()));
                    System.out.println("qualifier:" + Bytes.toString(kv.getQualifierArray()));
                    System.out.println("value:" + Bytes.toString(kv.getValueArray()));
                    System.out.println("timestamp:" + kv.getTimestamp());
                    System.out.println("-------------------------------------------");
                }
            }
        } finally {
            rs.close();
        }
    }

    /*
     * 查询表中的某一列
     *
     * @tableName 表名
     *
     * @rowKey rowKey
     */
    public void getResultByColumn(String tableName, String rowKey, String familyName, String columnName) throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName)); // 获取指定列族和列修饰符对应的列
        Result result = table.get(get);
        for (Cell kv : result.listCells()) {
            System.out.println("row:" + Bytes.toString(kv.getRowArray()));
            System.out.println("family:" + Bytes.toString(kv.getFamilyArray()));
            System.out.println("qualifier:" + Bytes.toString(kv.getQualifierArray()));
            System.out.println("value:" + Bytes.toString(kv.getValueArray()));
            System.out.println("timestamp:" + kv.getTimestamp());
            System.out.println("-------------------------------------------");
        }
    }

    /*
     * 更新表中的某一列
     *
     * @tableName 表名
     *
     * @rowKey rowKey
     *
     * @familyName 列族名
     *
     * @columnName 列名
     *
     * @value 更新后的值
     */
    public void updateTable(String tableName, String rowKey, String familyName, String columnName, String value)
            throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Put put = new Put(Bytes.toBytes(rowKey));
        put.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
        table.put(put);
        System.out.println("update table Success!");
    }

    /*
     * 查询某列数据的多个版本
     *
     * @tableName 表名
     *
     * @rowKey rowKey
     *
     * @familyName 列族名
     *
     * @columnName 列名
     */
    public void getResultByVersion(String tableName, String rowKey,
                                   String familyName, String columnName) throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
        get.setMaxVersions(5);
        Result result = table.get(get);
        for (Cell kv : result.listCells()) {
            System.out.println("family:" + Bytes.toString(kv.getFamilyArray()));
            System.out.println("qualifier:" + Bytes.toString(kv.getQualifierArray()));
            System.out.println("value:" + Bytes.toString(kv.getValueArray()));
            System.out.println("Timestamp:" + kv.getTimestamp());
            System.out.println("-------------------------------------------");
        }
        /*
         * List<?> results = table.get(get).list(); Iterator<?> it =
         * results.iterator(); while (it.hasNext()) {
         * System.out.println(it.next().toString()); }
         */
    }

    /*
     * 删除指定的列
     *
     * @tableName 表名
     *
     * @rowKey rowKey
     *
     * @familyName 列族名
     *
     * @columnName 列名
     */
    public void deleteColumn(String tableName, String rowKey, String falilyName, String columnName) throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Delete deleteColumn = new Delete(Bytes.toBytes(rowKey));
        deleteColumn.addColumn(Bytes.toBytes(falilyName), Bytes.toBytes(columnName));
        table.delete(deleteColumn);
        System.out.println(falilyName + ":" + columnName + "is deleted!");
    }

    /*
     * 删除指定的列
     *
     * @tableName 表名
     *
     * @rowKey rowKey
     */
    public void deleteAllColumn(String tableName, String rowKey)
            throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Table table = conn.getTable(_tableName);
        Delete deleteAll = new Delete(Bytes.toBytes(rowKey));
        table.delete(deleteAll);
        System.out.println("all columns are deleted!");
    }

    /*
     * 删除表
     *
     * @tableName 表名
     */
    public void deleteTable(String tableName) throws IOException {
        TableName _tableName = TableName.valueOf(tableName.getBytes());
        Admin admin = conn.getAdmin();
        admin.disableTable(_tableName);
        admin.deleteTable(_tableName);
        System.out.println(tableName + "is deleted!");
    }

    public static void main(String[] args) throws IOException {
        HBaseHolder hBaseHolder = new HBaseHolder();
        hBaseHolder.createTable(hBaseHolder.tableName);
        String[] column = {"title", "content", "tag"};
        String[] value = {
                "中文 First HBase",
                "HBase is the Hadoop database. Use it when you need random, realtime read/write access to your Big Data.",
                "Hadoop,HBase,NoSQL"};

        String[] column2 = {"name", "nickname"};

        String[] value2 = {
                "Head First HBase",
                "HBase is the Hadoop database. Use it when you need random, realtime read/write access to your Big Data."};

        String rowKey = "001";
        hBaseHolder.addData(rowKey, hBaseHolder.tableName, column, value, column2, value2);
        hBaseHolder.getResult(hBaseHolder.tableName, rowKey);
    }
}
